package gateway


import (
	"context"
	"crypto/tls"
	"fmt"
	"gitee.com/wayjin/fabric-extension/cryptoutil"
	_ "github.com/golang/protobuf/proto"
	commonutil "github.com/hyperledger/fabric/common/util"
	"github.com/hyperledger/fabric/core/comm"
	ccapi "github.com/hyperledger/fabric/peer/chaincode/api"
	peercommon "github.com/hyperledger/fabric/peer/common"
	"github.com/hyperledger/fabric/peer/common/api"
	cb "github.com/hyperledger/fabric/protos/common"
	ordererproto "github.com/hyperledger/fabric/protos/orderer"
	peerproto "github.com/hyperledger/fabric/protos/peer"
	"github.com/pkg/errors"
	"io/ioutil"
	"math"
	"sync"
	"time"
)

type commonClient struct {
	*comm.GRPCClient
	address string
	sn      string // service name
}

func newClientConfig(useTls bool, address, sn, tlsRootCertFile string) (clientConfig comm.ClientConfig, err error) {
	clientConfig = comm.ClientConfig{}
	secOpts := &comm.SecureOptions{
		UseTLS: useTls,
	}

	if secOpts.UseTLS {
		var caPem []byte
		caPem, err = ioutil.ReadFile(tlsRootCertFile)
		if err != nil {
			return
		}
		secOpts.ServerRootCAs = [][]byte{caPem}
	}
	clientConfig.Timeout = 3 * time.Second
	clientConfig.SecOpts = secOpts
	return
}

type ordererClient struct {
	commonClient
}

func newOrdererClient(address, sn, tlsRootCertFile string) (*ordererClient, error) {
	useTls := false
	if tlsRootCertFile != "" {
		useTls = true
	}

	config, err := newClientConfig(useTls, address, sn, tlsRootCertFile)
	if err != nil {
		return nil, err
	}
	gClient, err := comm.NewGRPCClient(config)
	if err != nil {
		return nil, err
	}
	return &ordererClient{commonClient{
		GRPCClient: gClient,
		address:    address,
		sn:         sn,
	}}, nil
}

func newOrdererClientWithTlsBytes(address, sn string, tlsRootBytes []byte) (*ordererClient, error) {
	clientConfig := comm.ClientConfig{}
	secOpts := &comm.SecureOptions{}
	if tlsRootBytes != nil {
		secOpts.UseTLS = true
		secOpts.ServerRootCAs = [][]byte{tlsRootBytes}
	}
	clientConfig.Timeout = 3 * time.Second
	clientConfig.SecOpts = secOpts
	gClient, err := comm.NewGRPCClient(clientConfig)
	if err != nil {
		return nil, err
	}
	return &ordererClient{commonClient{
		GRPCClient: gClient,
		address:    address,
		sn:         sn,
	}}, nil
}

func (oc *ordererClient) Broadcast() (ordererproto.AtomicBroadcast_BroadcastClient, error) {
	conn, err := oc.commonClient.NewConnection(oc.address, oc.sn)
	if err != nil {
		return nil, err
	}
	return ordererproto.NewAtomicBroadcastClient(conn).Broadcast(context.TODO())
}

func (oc *ordererClient) Deliver() (ordererproto.AtomicBroadcast_DeliverClient, error) {
	conn, err := oc.commonClient.NewConnection(oc.address, oc.sn)
	if err != nil {
		return nil, err
	}
	return ordererproto.NewAtomicBroadcastClient(conn).Deliver(context.TODO())
}

func (oc *ordererClient) Certificate() tls.Certificate {
	return oc.commonClient.Certificate()
}

func (oc *ordererClient) GetBroadcast() (BroadcastClient, error) {
	broadcastCli, err := oc.Broadcast()
	if err != nil {
		return nil, err
	}
	return &broadcastClient{broadcastCli}, nil
}

type ordererDelieverService struct {
	ordererproto.AtomicBroadcast_DeliverClient
}

func (oc *ordererClient) GetDeliverClient(channelID string, signer cryptoutil.Signer) (*ordererDeliverClient, error) {
	deliver, err := oc.Deliver()
	if err != nil {
		return nil, err
	}
	var tlsCertHash []byte
	if len(oc.Certificate().Certificate) > 0 {
		tlsCertHash = commonutil.ComputeSHA256(oc.Certificate().Certificate[0])
	}
	ds := &ordererDelieverService{deliver}
	return &ordererDeliverClient{
		Service:     ds,
		channelID:   channelID,
		tlsCertHash: tlsCertHash,
		signer:      signer,
	}, nil
}

type BroadcastClient interface {
	Send(env *cb.Envelope) error
	Close() error
}

type broadcastClient struct {
	client ordererproto.AtomicBroadcast_BroadcastClient
}

func (s *broadcastClient) getAck() error {
	msg, err := s.client.Recv()
	if err != nil {
		return fmt.Errorf("recv fail: %v", err)
	}
	if msg.Status != cb.Status_SUCCESS {
		return errors.Errorf("got unexpected status: %v -- %s", msg.Status, msg.Info)
	}
	return nil
}

func (s *broadcastClient) Send(env *cb.Envelope) error {
	if err := s.client.Send(env); err != nil {
		return errors.WithMessage(err, "could not send")
	}
	return s.getAck()
}

func (s *broadcastClient) Close() error {
	return s.client.CloseSend()
}

type ordererDeliverClient struct {
	Service     api.DeliverService
	channelID   string
	tlsCertHash []byte
	signer      cryptoutil.Signer
	bestEffort  bool
}

//
//func createSignedEnvelopeWithTLSBinding(txType cb.HeaderType, channelID string, signer Signer, dataMsg proto.Message, msgVersion int32, epoch uint64, tlsCertHash []byte) (*cb.Envelope, error) {
//	payloadChannelHeader := .MakeChannelHeader(txType, msgVersion, channelID, epoch)
//	payloadChannelHeader.TlsCertHash = tlsCertHash
//	var err error
//	payloadSignatureHeader := &cb.SignatureHeader{}
//
//	if signer != nil {
//		payloadSignatureHeader, err = signer.NewSignatureHeader()
//		if err != nil {
//			return nil, err
//		}
//	}
//
//	data, err := proto.Marshal(dataMsg)
//	if err != nil {
//		return nil, fmt.Errorf("error marshaling:%v", err)
//	}
//
//	paylBytes := utils.MarshalOrPanic(
//		&cb.Payload{
//			Header: utils.MakePayloadHeader(payloadChannelHeader, payloadSignatureHeader),
//			Data:   data,
//		},
//	)
//
//	var sig []byte
//	if signer != nil {
//		sig, err = signer.Sign(paylBytes)
//		if err != nil {
//			return nil, err
//		}
//	}
//
//	env := &cb.Envelope{
//		Payload:   paylBytes,
//		Signature: sig,
//	}
//	return env, nil
//}

func (dc ordererDeliverClient) seekHelper(pos *ordererproto.SeekPosition) (*cb.Envelope, error) {
	seekInfo := &ordererproto.SeekInfo{
		Start:    pos,
		Stop:     pos,
		Behavior: ordererproto.SeekInfo_BLOCK_UNTIL_READY,
	}

	return createSignedEnvelopeWithTLSBinding(
		cb.HeaderType_DELIVER_SEEK_INFO,
		dc.channelID,
		dc.signer,
		seekInfo,
		int32(0),
		uint64(0),
		dc.tlsCertHash)
}

func (dc ordererDeliverClient) seekSpecified(blockNum uint64) error {
	seekPos := &ordererproto.SeekPosition{
		Type: &ordererproto.SeekPosition_Specified{
			Specified: &ordererproto.SeekSpecified{
				Number: blockNum,
			},
		},
	}
	if dc.signer == nil {
		return fmt.Errorf("signer is nil")
	}
	env, err := dc.seekHelper(seekPos)
	if err != nil {
		return err
	}
	return dc.Service.Send(env)
}

func (dc ordererDeliverClient) getSpecifiedBlock(num uint64) (*cb.Block, error) {
	err := dc.seekSpecified(num)
	if err != nil {
		return nil, errors.WithMessage(err, "error getting specified block")
	}
	return dc.readBlock()
}

func (dc ordererDeliverClient) getMewestBlock() (*cb.Block, error) {
	newestPos := &ordererproto.SeekPosition{
		Type: &ordererproto.SeekPosition_Newest{
			Newest: &ordererproto.SeekNewest{},
		},
	}
	env, err := dc.seekHelper(newestPos)
	if err != nil {
		return nil, err
	}
	err = dc.Service.Send(env)
	if err != nil {
		return nil, fmt.Errorf("fail to send: %v", err)
	}

	return dc.readBlock()
}

func (dc ordererDeliverClient) getTopChainBlockByNum(topCount uint64) ([]*cb.Block, error) {
	newest, err  := dc.getMewestBlock()
	if err != nil {
		fmt.Println("error:", err)
		return nil, err
	}

	if topCount > newest.Header.Number {
		topCount = newest.Header.Number
	}
	blocks := make([]*cb.Block, topCount)
	fmt.Println("newest block number:", newest.Header.Number)
	blocks[0] = newest
	end := newest.Header.Number - topCount
	j := 1
	for i:=newest.Header.Number-1; i>end; i-- {
		fmt.Println("get block by number:", i)
		b, err := dc.getSpecifiedBlock(i)
		if err != nil {
			continue
		}
		blocks[j] = b
		j++
	}
	return blocks, nil
}

func (dc ordererDeliverClient) readBlock() (*cb.Block, error) {
	msg, err := dc.Service.Recv()
	if err != nil {
		return nil, errors.Wrap(err, "error receiving")
	}
	switch t := msg.Type.(type) {
	case *ordererproto.DeliverResponse_Status:
		return nil, errors.Errorf("can't read the block: %v", t)
	case *ordererproto.DeliverResponse_Block:
		dc.Service.Recv() // Flush the success message
		return t.Block, nil
	default:
		return nil, errors.Errorf("response error: unknown type %T", t)
	}
}

func (dc ordererDeliverClient) Close() {
	dc.Service.CloseSend()
}

type peerClient struct {
	commonClient
}

func newPeerCient(address, sn, tlsRootCertFile string) (*peerClient, error) {
	useTls := tlsRootCertFile != ""
	config, err := newClientConfig(useTls, address, sn, tlsRootCertFile)
	if err != nil {
		return nil, err
	}
	gClient, err := comm.NewGRPCClient(config)
	if err != nil {
		return nil, err
	}
	return &peerClient{commonClient{
		GRPCClient: gClient,
		address:    address,
		sn:         sn,
	}}, nil
}

func newPeerClientWithTLSBytes(address, sn string, tlsRootBytes []byte) (*peerClient, error) {
	clientConfig := comm.ClientConfig{}
	secOpts := &comm.SecureOptions{}
	if tlsRootBytes != nil {
		secOpts.UseTLS = true
		secOpts.ServerRootCAs = [][]byte{tlsRootBytes}
	}
	clientConfig.Timeout = 3 * time.Second
	clientConfig.SecOpts = secOpts
	gClient, err := comm.NewGRPCClient(clientConfig)
	if err != nil {
		return nil, err
	}
	return &peerClient{commonClient{
		GRPCClient: gClient,
		address:    address,
		sn:         sn,
	}}, nil
}

func (pc *peerClient) Endorser() (peerproto.EndorserClient, error) {
	conn, err := pc.commonClient.NewConnection(pc.address, pc.sn)
	if err != nil {
		return nil, errors.WithMessage(err, fmt.Sprintf("endorser client failed to connect to %s", pc.address))
	}
	return peerproto.NewEndorserClient(conn), nil
}

func (pc *peerClient) Deliver() (peerproto.Deliver_DeliverClient, error) {
	conn, err := pc.commonClient.NewConnection(pc.address, pc.sn)
	if err != nil {
		return nil, errors.WithMessage(err, fmt.Sprintf("deliver client failed to connect to %s", pc.address))
	}
	return peerproto.NewDeliverClient(conn).Deliver(context.TODO())
}

func (pc *peerClient) PeerDeliver() (api.PeerDeliverClient, error) {
	conn, err := pc.commonClient.NewConnection(pc.address, pc.sn)
	if err != nil {
		return nil, errors.WithMessage(err, fmt.Sprintf("deliver client failed to connect to %s", pc.address))
	}
	peerCli := peerproto.NewDeliverClient(conn)
	return &peercommon.PeerDeliverClient{Client: peerCli}, nil
}

func (pc *peerClient) processProposal(signedProp *peerproto.SignedProposal) (*peerproto.ProposalResponse, error) {
	endorser, err := pc.Endorser()
	if err != nil {
		return nil, err
	}
	return endorser.ProcessProposal(context.Background(), signedProp)
}

func (pc *peerClient) getAdmin() (peerproto.AdminClient, error) {
	gCli, err := pc.commonClient.NewConnection(pc.address, pc.sn)
	if err != nil {
		return nil, err
	}
	return peerproto.NewAdminClient(gCli), nil
}

type deliverGroup struct {
	Clients     []*peerDelieverClient
	Certificate tls.Certificate
	ChannelID   string
	TxID        string
	signer      cryptoutil.Signer // add
	mutex       sync.Mutex
	Error       error
	wg          sync.WaitGroup
}

type peerDelieverClient struct {
	Client     api.PeerDeliverClient
	Connection ccapi.Deliver
	Address    string
}

func newDeliverGroup(deliverClients []api.PeerDeliverClient, peerAddresses []string, certificate tls.Certificate, signer cryptoutil.Signer, channelID string, txid string) *deliverGroup {
	clients := make([]*peerDelieverClient, len(deliverClients))
	for i, client := range deliverClients {
		dc := &peerDelieverClient{
			Client:  client,
			Address: peerAddresses[i],
		}
		clients[i] = dc
	}

	dg := &deliverGroup{
		Clients:     clients,
		Certificate: certificate,
		ChannelID:   channelID,
		TxID:        txid,
		signer:      signer,
	}

	return dg
}

// Connect waits for all deliver clients in the group to connect to
// the peer's deliver service, receive an error, or for the context
// to timeout. An error will be returned whenever even a single
// deliver client fails to connect to its peer
func (dg *deliverGroup) Connect(ctx context.Context) error {
	dg.wg.Add(len(dg.Clients))
	for _, client := range dg.Clients {
		go dg.ClientConnect(ctx, client)
	}
	readyCh := make(chan struct{})
	go dg.WaitForWG(readyCh)

	select {
	case <-readyCh:
		if dg.Error != nil {
			err := errors.WithMessage(dg.Error, "failed to connect to deliver on all peers")
			return err
		}
	case <-ctx.Done():
		err := errors.New("timed out waiting for connection to deliver on all peers")
		return err
	}

	return nil
}

// ClientConnect sends a deliver seek info envelope using the
// provided deliver client, setting the deliverGroup's Error
// field upon any error
func (dg *deliverGroup) ClientConnect(ctx context.Context, dc *peerDelieverClient) {
	defer dg.wg.Done()
	df, err := dc.Client.DeliverFiltered(ctx)
	if err != nil {
		err = errors.WithMessage(err, fmt.Sprintf("error connecting to deliver filtered at %s", dc.Address))
		dg.setError(err)
		return
	}
	defer df.CloseSend()
	dc.Connection = df

	envelope := createDeliverEnvelope(dg.ChannelID, dg.Certificate, dg.signer)
	if envelope == nil {
		err = errors.Errorf("empty envelope")
		dg.setError(err)
		return
	}
	err = df.Send(envelope)
	if err != nil {
		err = errors.WithMessage(err, fmt.Sprintf("error sending deliver seek info envelope to %s", dc.Address))
		dg.setError(err)
		return
	}
}

// setError serializes an error for the deliverGroup
func (dg *deliverGroup) setError(err error) {
	dg.mutex.Lock()
	dg.Error = err
	dg.mutex.Unlock()
}

// WaitForWG waits for the deliverGroup's wait group and closes
// the channel when ready
func (dg *deliverGroup) WaitForWG(readyCh chan struct{}) {
	dg.wg.Wait()
	close(readyCh)
}

func createDeliverEnvelope(channelID string, certificate tls.Certificate, signer cryptoutil.Signer) *cb.Envelope {
	var tlsCertHash []byte
	// check for client certificate and create hash if present
	if len(certificate.Certificate) > 0 {
		tlsCertHash = commonutil.ComputeSHA256(certificate.Certificate[0])
	}

	start := &ordererproto.SeekPosition{
		Type: &ordererproto.SeekPosition_Newest{
			Newest: &ordererproto.SeekNewest{},
		},
	}

	stop := &ordererproto.SeekPosition{
		Type: &ordererproto.SeekPosition_Specified{
			Specified: &ordererproto.SeekSpecified{
				Number: math.MaxUint64,
			},
		},
	}

	seekInfo := &ordererproto.SeekInfo{
		Start:    start,
		Stop:     stop,
		Behavior: ordererproto.SeekInfo_BLOCK_UNTIL_READY,
	}

	//env, err := protoutils.createSignedEnvelopeWithTLSBinding(
	env, err := createSignedEnvelopeWithTLSBinding(
		cb.HeaderType_DELIVER_SEEK_INFO, channelID, signer, //localmsp.NewSigner(),
		seekInfo, int32(0), uint64(0), tlsCertHash)
	if err != nil {
		fmt.Println("Error signing envelope: ", err)
		return nil
	}

	return env
}

// Wait waits for all deliver client connections in the group to
// either receive a block with the txid, an error, or for the
// context to timeout
func (dg *deliverGroup) Wait(ctx context.Context) error {
	if len(dg.Clients) == 0 {
		return nil
	}

	dg.wg.Add(len(dg.Clients))
	for _, client := range dg.Clients {
		go dg.ClientWait(client)
	}
	readyCh := make(chan struct{})
	go dg.WaitForWG(readyCh)

	select {
	case <-readyCh:
		if dg.Error != nil {
			err := errors.WithMessage(dg.Error, "failed to receive txid on all peers")
			return err
		}
	case <-ctx.Done():
		err := errors.New("timed out waiting for txid on all peers")
		return err
	}

	return nil
}

// ClientWait waits for the specified deliver client to receive
// a block event with the requested txid
func (dg *deliverGroup) ClientWait(dc *peerDelieverClient) {
	defer dg.wg.Done()
	for {
		resp, err := dc.Connection.Recv()
		if err != nil {
			err = errors.WithMessage(err, fmt.Sprintf("error receiving from deliver filtered at %s", dc.Address))
			dg.setError(err)
			return
		}
		switch r := resp.Type.(type) {
		case *peerproto.DeliverResponse_FilteredBlock:
			filteredTransactions := r.FilteredBlock.FilteredTransactions
			for _, tx := range filteredTransactions {
				if tx.Txid == dg.TxID {
					//logger.Infof("txid [%s] committed with status (%s) at %s", dg.TxID, tx.TxValidationCode, dc.Address)
					return
				}
			}
		case *peerproto.DeliverResponse_Status:
			err = errors.Errorf("deliver completed with status (%s) before txid received", r.Status)
			dg.setError(err)
			return
		default:
			err = errors.Errorf("received unexpected response type (%T) from %s", r, dc.Address)
			dg.setError(err)
			return
		}
	}
}
